package cn.getech.data.development.utils

import java.sql.{Connection, Statement}

import cn.getech.data.development.bean.jdbc.{JDBCFieldBean, JDBCSqlParserResourceBean}
import cn.getech.data.development.enums.FlinkStreamHiveWriteMode
import org.slf4j.{Logger, LoggerFactory}

object HiveUtils {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass)
  var obj: JDBCSqlParserResourceBean = null
  var connection: Connection = null
  var stmt: Statement = null

  def open(obj: JDBCSqlParserResourceBean): Unit = {
    this.obj = obj
    obj.hiveWriteMode match {
      case FlinkStreamHiveWriteMode.APPEND =>
        logger.info("hive write mode is append！")
      case FlinkStreamHiveWriteMode.COVER =>
        logger.info("hive write mode is cover！")
        setCover()
    }
  }

  def partitionPath(partitions: Array[JDBCFieldBean], delimit: String = "/"): String = {
    if (obj.isDynamicPartition) return ""
    partitions.filter(_.isPartition).map(x => {
      if (x.partitionValue.startsWith("${") && x.partitionValue.endsWith("}")) {
        val format = x.partitionValue.replace("${", "").replace("}", "")
        val utils = new DateUtils()
        s"${x.field_name}=${utils.getPatternTime(format)}"
      }
      else s"${x.field_name}=${x.partitionValue}"
    }).mkString(delimit)
  }

  def partitionSQLPath(partitions: Array[JDBCFieldBean], delimit: String = "/"): String = {
    partitions.filter(_.isPartition).map(x => s"${x.field_name}='${x.partitionValue}'").mkString(delimit)
  }

  private def setCover(): Unit = {
    if (obj.isPartitionTable) {
      val value = s"${ConfigurationManager.getProperty("hdfs.hive.warehouse.path")}${obj.db}.db/${obj.tableName}/"
      val parPath = partitionPath(obj.fields)
      logger.info(s"hdfs path: $value$parPath")
      HDFSUtils.overridePath(value, parPath)
    } else {
      val value = s"${ConfigurationManager.getProperty("hdfs.hive.warehouse.path")}${obj.db}.db/"
      HDFSUtils.overridePath(value, obj.tableName)
    }
  }

  @Deprecated
  private def addPartitions(): Boolean = {
    if (stmt == null) return false
    stmt.execute(s"ALTER TABLE ${obj.db}.${obj.tableName} add if not exists PARTITION (${partitionSQLPath(obj.fields, ",")})")
  }

  @Deprecated
  private def dropPartitions(): Boolean = {
    if (stmt == null) return false
    stmt.execute(s"ALTER TABLE ${obj.db}.${obj.tableName} drop if exists PARTITION (${partitionSQLPath(obj.fields, ",")})")
  }

  @Deprecated
  private def externalPartitions(external: String): Boolean = {
    if (stmt == null) return false
    stmt.execute(s"ALTER TABLE ${obj.db}.${obj.tableName} SET TBLPROPERTIES('EXTERNAL'='${external.toUpperCase}')")
  }

  @Deprecated
  private def truncateTable: Boolean = {
    if (stmt == null) return false
    stmt.execute(s"truncate table ${obj.db}.${obj.tableName}")
  }

  @Deprecated
  private def close: Unit = {
    if (stmt != null) {
      stmt.close()
    }

    if (connection != null) {
      connection.close
    }
  }
}
